package com.mattgarner.jaddas.dataset;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import com.mattgarner.jaddas.node.DataSetManager;
import com.mattgarner.jaddas.node.data.LocalTable;
import com.mattgarner.jaddas.node.data.TableException;
import com.mattgarner.jaddas.node.index.ColumnIndexStatistics;
import com.mattgarner.jaddas.node.index.LocalIndex;

public class DataSetPartition {

	private DataSet dataSet;
	
	private LocalTable localTable;
	private LocalIndex localIndex;
	
	private int partitionID;
	
	private int nextTupleID = -1;
	
	private int dspSerialNumber;
	private int dspDataSerialNumber;
	
	private boolean isLocal;
	
	private ColumnIndexStatistics[] colIndexStats;
	
	private DataSetManager dsManager;
	
	public DataSetPartition(DataSet dataSet, int partitionID, boolean isLocal) {
		this.dataSet = dataSet;
		this.partitionID = partitionID;
		this.isLocal = isLocal;
		
		int columnCount = 0;
		try {
			columnCount = dataSet.getDataTableHeader().getColumnCount();
			
			// setup some empty column statistics
			this.colIndexStats = new ColumnIndexStatistics[columnCount];
			for(int a=0; a<this.colIndexStats.length; a++) {
				this.colIndexStats[a] = new ColumnIndexStatistics();
				this.colIndexStats[a].setColumnID(a);
				this.colIndexStats[a].setColumnType(dataSet.getDataTableHeader().getColumnBasicType(a));
			}
			
		} catch(TableException e) {
			// TODO: something should be done here 
		}
		
		
		this.dsManager = DataSetManager.getInstance();
	}
	
	public DataSetPartition(DataSet dataSet, byte[] serializedBytes) {
		this.dataSet = dataSet; 
		initializeFromByteArray(serializedBytes);
		
		this.dsManager = DataSetManager.getInstance();
	}
	
	// getters 
	
	public final int getPartitionID() {
		return this.partitionID;
	}
	
	public final boolean isLocal() {
		return this.isLocal;
	}
	
	public final int getCurrentTupleID() {
		return this.nextTupleID;
	}
	
	public final int getNextTupleID() {
		dspSerialNumber++;
		this.nextTupleID++;
		dsManager.saveDataSetPartitionTupleIDToDisk(this);	
		return this.nextTupleID;
	}
	
	public final DataSet getParentDataSet() {
		return this.dataSet;
	}
	
	public final LocalTable getLocalTable() {
		return this.localTable;
	}
	
	public final LocalIndex getLocalIndex() {
		return this.localIndex;
	}
	
	public final ColumnIndexStatistics getColumnIndexStatistic(int column) {
		return colIndexStats[column];
	}

	public final ColumnIndexStatistics[] getAllColumnIndexStatistics() {
		return colIndexStats;
	}
	
	public final boolean isOpen() {
		if(dataSet.getOpenPartitionID() == this.partitionID) {
			return true;
		} else {
			return false;
		}
	}
	
	// setters
	
	public final void setLocalTable(final LocalTable localTable) {
		this.localTable = localTable;
	}
	
	public final void setLocalIndex(final LocalIndex localIndex) {
		this.localIndex = localIndex;
	}
	
	public final void setCurrentTupleID(int tupleID) {
		this.nextTupleID = tupleID;
	}
	
	public final void setColumnIndexStatistic(int columnID, ColumnIndexStatistics colIndexStat) {
		this.colIndexStats[columnID] = colIndexStat;
	}
	
	private void initializeFromByteArray(byte[] serializedBytes) {
		ByteArrayInputStream bais = new ByteArrayInputStream(serializedBytes);
		DataInputStream stream = new DataInputStream(bais);
		
		try {
			partitionID = stream.readInt();
			isLocal = stream.readBoolean();
			dspSerialNumber = stream.readInt();
			dspDataSerialNumber = stream.readInt();
			
			// read an array of ColumnIndexStatistics bytes 
			int columnCount = dataSet.getDataTableHeader().getColumnCount();
			colIndexStats = new ColumnIndexStatistics[columnCount];
			for(int a=0; a<columnCount; a++) {
				int cStatsBytesLength = stream.readInt();
				byte[] cStatsBytes = new byte[cStatsBytesLength];
				stream.read(cStatsBytes, 0, cStatsBytesLength);
				colIndexStats[a] = new ColumnIndexStatistics(cStatsBytes);	
			}
			

			stream.close();
			bais.close();
		} catch(Exception e) {
			return;
		}
	}
	
	public final byte[] serializeToByteArray() {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream stream = new DataOutputStream(baos);
				
		try {
			stream.writeInt(partitionID);
			stream.writeBoolean(isLocal);
			stream.writeInt(dspSerialNumber);
			stream.writeInt(dspDataSerialNumber);
			
			for(int a = 0; a<colIndexStats.length; a++) {
				byte[] colIndexStatsBytes = colIndexStats[a].serializeToByteArray();
				stream.writeInt(colIndexStatsBytes.length);
				stream.write(colIndexStatsBytes);
			}
			
			stream.close();
			baos.close();
			return baos.toByteArray();
		} catch (IOException e) {
			return null;
		}
	}
}
